package com.lecosa.flink;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.flink.configuration.Configuration;

public class ChangeSink extends RichSinkFunction<String> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private String lastValue = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载驱动程序
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 获取数据库连接
        String url = "jdbc:mysql://192.168.249.112:3306/shufen";
        String user = "root";
        String password = "root";
        connection = DriverManager.getConnection(url, user, password);
        // 创建 PreparedStatement 对象
        String sql = "replace INTO student1 (id,no, name, gender) VALUES (?,?, ?, ?)";
        preparedStatement = connection.prepareStatement(sql);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 如果上一次处理的值为空，或者当前值与上一次处理的值不相同，则执行 sink 操作
        if (lastValue == null || !lastValue.equals(value)) {
            // 执行 sink 操作
            String id = "";
            String name = "";
            String no = "";
            String gender = "";

            // 匹配 id 的正则表达式
            Pattern idPattern = Pattern.compile("id=(\\d+)");
            Matcher idMatcher = idPattern.matcher(value);
            if (idMatcher.find()) {
                id = idMatcher.group(1);
            }

            // 匹配 no 的正则表达式
            Pattern noPattern = Pattern.compile("no=([^,]+)");
            Matcher noMatcher = noPattern.matcher(value);
            if (noMatcher.find()) {
                no = noMatcher.group(1);
            }

            // 匹配 name 的正则表达式
            Pattern namePattern = Pattern.compile("name=([^,]+)");
            Matcher nameMatcher = namePattern.matcher(value);
            if (nameMatcher.find()) {
                name = nameMatcher.group(1);
            }

            // 匹配 gender 的正则表达式
            Pattern genderPattern = Pattern.compile("gender=([^,}]+)");
            Matcher genderMatcher = genderPattern.matcher(value);
            if (genderMatcher.find()) {
                gender = genderMatcher.group(1);
            }

            // 将值插入到新表中
            preparedStatement.setString(1, id);
            preparedStatement.setString(2, no);
            preparedStatement.setString(3, name);
            preparedStatement.setString(4, gender);
            preparedStatement.executeUpdate();

            // 更新上一次处理的值为当前值
            lastValue = value;
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
